package com.bw.gmall.realtime.app.dws;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwsShopUserPay {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        String topic ="dwd_trade_pay_suc_province";
        String groupId ="DwsShopUserPay_2116";
        tenv.executeSql("CREATE TABLE DwsShopUserPay (\n" +
                "  `shop_id` STRING,\n" +
                "  `province_name` STRING,\n" +
                "  `user_id` STRING,\n" +
                "  `row_op_ts` STRING,\n" +  // 注意：这里是字符串类型
                "  `time_ltz` AS TO_TIMESTAMP_LTZ(\n" +
                "    UNIX_TIMESTAMP(REPLACE(row_op_ts, ' ', 'T'), 'yyyy-MM-dd''T''HH:mm:ss.SSSX') * 1000,\n" +
                "    3\n" +  // 毫秒精度
                "  ),\n" +
                "  WATERMARK FOR time_ltz AS time_ltz\n" +
                ")" + MyKafkaUtil.getKafkaDDL(topic, groupId));

        tenv.executeSql("SELECT *\n" +
                "  FROM (" +
                "SELECT *, ROW_NUMBER() OVER (PARTITION BY window_start, window_end ORDER BY unum DESC) as rownum\n" +
                "    FROM (\n" +
                "      SELECT window_start, window_end, shop_id,province_name, count(distinct user_id) as unum, COUNT(*) as paynum\n" +
                "      FROM TABLE(\n" +
                "        TUMBLE(TABLE DwsShopUserPay, DESCRIPTOR(time_ltz), INTERVAL '1' seconds))\n" +
                "      GROUP BY window_start, window_end, shop_id,province_name\n" +
                "    )\n" +
                ")   WHERE rownum <= 10 ").print();



        env.execute();
    }
}
